# -*- coding: utf-8 -*-

"""
@author: Laowang
@contact: QQ:1125564921
@Created on: 2022/6/20
@Remark: 
"""
import os
import socket
import random
from select import select
from queue import Queue
from threading import Event, Thread, Lock, Condition


class ActorExit(Exception):
    ...

# PollableQueue


class Result:

    def __init__(self) -> None:
        self._lock = Lock()
        self._value = []
        self._e = Event()

    def set_value(self, value):
        with self._lock:
            self._value.append(value)
        self._e.set()

    def result(self):
        self._e.wait()
        return self._value

    def __iter__(self):
        return iter(self._value)


class PollableQueue(Queue):
    """ Deal with excess CPU waste """

    def __init__(self, maxsize: int = 0) -> None:

        super().__init__(maxsize)
        if os.name == "posix":
            self._putsocket, self._getsocket = socket.socketpair()
        else:
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.bind(("localhost", 0))
            server.listen(1)

            self._putsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._putsocket.connect(server.getsockname())
            self._getsocket, _ = server.accept()

    def put(self, item, block=True, timeout=None):
        super().put(item, block, timeout)
        self._putsocket.send(b"x")

    def get(self, block=True, timeout=None):
        self._getsocket.recv(1)
        return super().get(block, timeout)

    def fileno(self):
        return self._getsocket.fileno()


class Actor:
    """ """

    def __init__(self, error=ActorExit, name="Actors") -> None:
        """ """

        self._error = error
        self._queue = Queue()
        self._event = Event()
        self._t = Thread(target=self._bootstrap, daemon=True, name=name)

    def recv(self, **kw):
        """ """
        item = self.get(**kw)
        if item is self._error:
            raise self._error()
        return item

    def get(self, **kw):
        """ """
        self._event.wait()
        return self._queue.get(**kw)

    def put(self, item, **kw):
        """ """
        self._queue.put(item, **kw)

    def send(self, item, **kw):
        """ """
        self.put(item, **kw)
        self._event.set()

    def _bootstrap(self):
        """ """
        while True:
            try:
                self.recv()
            except self._error:
                break
            except Exception:
                pass
            finally:
                self._event.set()

    def start(self):
        """ """
        self._t.start()

    def stop(self):
        """ """
        self.put(self._error)

    def join(self):
        """ """
        self._event.wait()


class MultiActors:
    """ Raise fan in and fan out """
    MAX_SIZE = 10

    def __init__(self, error=ActorExit, name="MultiActors") -> None:

        self._error = error
        self._queues = [PollableQueue() for _ in range(self.MAX_SIZE)]
        self._t = Thread(target=self._run, daemon=True, name=name)

        self._lock = Lock()
        self._put_mutex = Condition(self._lock)

    def recv(self, **kw):
        """ """
        for item in self.get(**kw):
            yield item

    def get(self, **kw):
        """ """
        res = Result()
        can_reads, _, _ = select(self._queues, [], [])
        for q in can_reads:
            res.set_value(q.get(**kw))
        yield from res

    def put(self, item, **kw):
        """ """
        with self._put_mutex:
            queue = random.choice(self._queues)
        queue.put(item, **kw)

    def send(self, item, **kw):
        """ """
        self.put(item, **kw)

    def _run(self):

        while True:
            try:
                self._bootstrap()
            except self._error:
                break
        print("[EventEngine] exit ...")

    def run(self):
        for item in self.recv():
            if item is self._error:
                raise self._error()
            print("[recv] ", item)

    def _bootstrap(self):
        """ """
        self.run()

    def start(self):
        """ """
        self._t.start()

    def stop(self):
        """ """
        self.put(self._error)

    def join(self):
        """ """
        self._t.join()

